import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/*作用：
* 1.把三张表写入MongoDB里
* 2.把三张表写入ES里【暂时不管】*/
object DataLoader extends App {
  //环境初始化（系统环境初始化 (SparkConf ==》Spark Session ==>sc)）
  private val spark: SparkSession = Constant.initEnv()


  //2.加载本地数据[RDD] （一种：SparkCore（sc）==>sc.TextFile
  // 二种：SparkSQL==>spark:SparkSession(内部封装了sparkcontext)  spark.sperkContext ==>ssc）
  private val movieRDD: RDD[String] = spark.sparkContext.textFile(Constant.MOVIE_PATH)//"电影表"
  private val ratingRDD: RDD[String] = spark.sparkContext.textFile(Constant.RATING_PATH)//"评分表"
  private val tagRDD: RDD[String] = spark.sparkContext.textFile(Constant.TAG_PATH)//"标签表"


  //3.RDD ==转换==>DataFrame(隐式转换导包) (1,"yyb") ==>teacher(1,"yyb")
  import  spark.implicits._
  /*电影表数据处理方式*/
  private val movieDF: DataFrame = movieRDD.map {
    item => {
      //按行读取按 ^ 分割
      val attr = item.split("\\^")
      //调用电影标签表Movie类 注：trim()的作用是去掉字bai符串两端的多余du的空格
      Movie(
        attr(0).trim.toInt,
        attr(1).trim,
        attr(2).trim,
        attr(3).trim,
        attr(4).trim,
        attr(5).trim,
        attr(6).trim,
        attr(7).trim,
        attr(8).trim,
        attr(9).trim

      )
    }

  }.toDF()

  /*评分表数据相同处理*/
  private val ratingDF: DataFrame = ratingRDD.map {
    item => {
      val attr = item.split(",")
      MovieRating(
        attr(0).toInt,
        attr(1).toInt,
        attr(2).toDouble,
        attr(3).toLong
      )
    }
  }.toDF
  println(ratingDF.show())
  /*标签表数据相同处理*/
  private val tagDF: DataFrame =tagRDD.map {
    item => {
      val attr = item.split(",")
      Tag(
        //注：（0）对应样例类的属性顺序
        attr(0).trim.toInt,
        attr(1).trim.toInt,
        attr(2).trim,
        attr(3).trim.toLong
      )
    }
  }.toDF

  //4.隐式声明连接到MongDB 地址，库（调用自己的样例类MongoConfig()函数）
  implicit private val mongoConfig: MongoConfig = MongoConfig(Constant.MONGO_URL,Constant.MONGO_DB)

  //5.调用写入MongoDB的方法
  storeDataIntoMongoDB(movieDF,ratingDF,tagDF)

  /*创建存入MongoDB的自定义函数，
  参数为：三张数据表,
         隐式转换的MongoDB连接配置*/
  def storeDataIntoMongoDB(
                            movieDF: DataFrame,
                            ratingDF: DataFrame,
                            tagDF: DataFrame)(implicit mongoConfig: MongoConfig) = {

   //将数据（已经转换成了DF） 写入MongoDB的具体过程
   //1.创建客户端
   val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))

   //2.查询到有表名/collection该就删掉,没有则创建
    // ( mongoClient("库名")("表名/collection").dropCollection())
    mongoClient(mongoConfig.db)(Constant.MOVIE_COLLECTION).dropCollection()
    mongoClient(mongoConfig.db)(Constant.RATING_COLLECTION).dropCollection()
    mongoClient(mongoConfig.db)(Constant.TAG_COLLECTION).dropCollection()



   //3.写入MongoDB（由于三张表写入MongoDB的语法步骤相同造成冗余重复
    // 则需将他们的过程封装起来在Constant类里）
    // df:DataFrame,collectionName:String 不同的地方
   /* movieDF.write
      .option("uri",mongoConfig.uri)
      .option("collection",Constant.MOVIE_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()*/
    Constant.DataIntoMongoDB(movieDF,Constant.MOVIE_COLLECTION)//Movie表
    Constant.DataIntoMongoDB(ratingDF,Constant.RATING_COLLECTION)//Rating表
    Constant.DataIntoMongoDB(tagDF,Constant.TAG_COLLECTION)//Tag表

   //4.创建索引 (类似id==三张表里一共有多少id就创建多少索引)
    mongoClient(mongoConfig.db)(Constant.MOVIE_COLLECTION).createIndex(MongoDBObject("mid"->1))//mid
    mongoClient(mongoConfig.db)(Constant.RATING_COLLECTION).createIndex(MongoDBObject("uid"->1))//uid mid
    mongoClient(mongoConfig.db)(Constant.RATING_COLLECTION).createIndex(MongoDBObject("mid"->1))
    mongoClient(mongoConfig.db)(Constant.TAG_COLLECTION).createIndex(MongoDBObject("uid"->1))//uid mid
    mongoClient(mongoConfig.db)(Constant.TAG_COLLECTION).createIndex(MongoDBObject("mid"->1))

  }
}
